package com.gitee.xhs.window.lateness;

import com.gitee.xhs.OperatorSourceFactory;
import com.gitee.xhs.customFunction.OperatorKeyBy;
import com.gitee.xhs.customWaterMark.CustomWaterStrategy;
import com.gitee.xhs.dao.AggregateTransactionDao;
import com.gitee.xhs.window.windowOperator.DefaultAggregateWindowFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import static com.gitee.xhs.OperatorConstant.LATENESS_OUT_PUT_TAG;

/**
 * @author BigHead
 * @date 2022/7/27 17:21
 * @description SimpleLatenessMain
 */
public class SimpleLatenessMain {

    private final StreamExecutionEnvironment streamExecutionEnvironment;

    public SimpleLatenessMain(StreamExecutionEnvironment streamExecutionEnvironment) {
        this.streamExecutionEnvironment = streamExecutionEnvironment;
    }

    public void latenessLogic() {
        SingleOutputStreamOperator<AggregateTransactionDao> allowLateDataStream = streamExecutionEnvironment
                .addSource(
                        OperatorSourceFactory.unBoundLoopOneToTenOneSecondSource()
                )
                .flatMap(
                        new OperatorLateDataFlatMap()
                )
                .assignTimestampsAndWatermarks(
                        new CustomWaterStrategy()
                )
                .keyBy(
                        new OperatorKeyBy()
                )
                .window(
                        TumblingEventTimeWindows.of(Time.minutes(1))
                )
                .allowedLateness(
                        Time.seconds(50)
                )
                .sideOutputLateData(
                        LATENESS_OUT_PUT_TAG
                )
                .aggregate(
                        new DefaultAggregateWindowFunction()
                );

        allowLateDataStream
                .print("非迟到数据 效果");

        allowLateDataStream
                .getSideOutput(
                        LATENESS_OUT_PUT_TAG
                )
                .print("迟到数据 效果");

    }

}
